package com.lz.ad.canal.processor;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.gson.Gson;
import com.lz.ad.canal.common.Config;
import com.lz.ad.canal.common.Constant;
import com.lz.ad.canal.config.kafka.KafkaProducer;
import com.lz.common.model.canal.RowChangeInfo;
import com.lz.common.util.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.SystemUtils;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.slf4j.MDC;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.net.InetSocketAddress;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by Li Tiefei on 2018/11/26.
 */
@Component
@Slf4j
public class CanalProcessor implements InitializingBean, DisposableBean,Runnable {

    private Thread thread;
    private volatile boolean running = false;
    private CanalConnector connector;
    private String destination;
    private static String context_format = null;
    private static String row_format = null;
    private static String transaction_format = null;

    private static final String SEP = SystemUtils.LINE_SEPARATOR;
    private static final String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";
    private static final Gson GSON = new Gson();

    protected Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() {

        public void uncaughtException(Thread t, Throwable e) {
            log.error("parse events has an error", e);
        }
    };

    @Autowired
    private KafkaProducer kafkaProducer;

    static {
        context_format = SEP + "****************************************************" + SEP;
        context_format += "* Batch Id: [{}] ,count : [{}] , memsize : [{}] , Time : {}" + SEP;
        context_format += "* Start : [{}] " + SEP;
        context_format += "* End : [{}] " + SEP;
        context_format += "****************************************************" + SEP;

        row_format = SEP
                + "----------------> binlog[{}:{}] , name[{},{}] , eventType : {} , executeTime : {} , delay : {}ms"
                + SEP;

        transaction_format = SEP + "================> binlog[{}:{}] , executeTime : {} , delay : {}ms" + SEP;
    }

    CanalProcessor() {
        this.thread = new Thread(this);
        Config.init();
        // canal 的 ip、port
        String ip = StringUtils.isNotBlank(Config.CANAL_IP) ? Config.CANAL_IP : AddressUtils.getHostIp();
        int port = Config.CANAL_PORT;
        // lzmhad Canal
        destination = Constant.DESTINATION_LZMHAD;
        // 创建canal连接
        CanalConnector connectorLzmhad = CanalConnectors.newSingleConnector(new InetSocketAddress(ip, port),
                destination, "", "");
        this.connector=connectorLzmhad;
        this.running=true;
        this.thread.setUncaughtExceptionHandler(handler);
        this.thread.start();
    }

    @Override
    public void run() {
        int batchSize = 5 * 1024;
        while (running) {
            try {
                MDC.put("destination", destination);
                connector.connect();
                connector.subscribe();
                while (running) {
                    Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                    long batchId = message.getId();
                    int size = message.getEntries().size();
                    if (batchId == -1 || size == 0) {
                         try {
                            Thread.sleep(1000);
                         } catch (InterruptedException e) {
                             e.printStackTrace();
                         }
                    } else {
                        printSummary(message, batchId, size);
                        printEntry(message.getEntries());
                    }

                    connector.ack(batchId); // 提交确认
//                    connector.rollback(batchId); // 处理失败, 回滚数据
                }
            } catch (Exception e) {
                log.error("process error!", e);
            } finally {
                connector.disconnect();
                MDC.remove("destination");
            }
        }
    }

    private String buildPositionForDump(CanalEntry.Entry entry) {
        long time = entry.getHeader().getExecuteTime();
        Date date = new Date(time);
        SimpleDateFormat format = new SimpleDateFormat(DATE_FORMAT);
        return entry.getHeader().getLogfileName() + ":" + entry.getHeader().getLogfileOffset() + ":"
                + entry.getHeader().getExecuteTime() + "(" + format.format(date) + ")";
    }

    private void printSummary(Message message, long batchId, int size) {
        long memsize = 0;
        for (CanalEntry.Entry entry : message.getEntries()) {
            memsize += entry.getHeader().getEventLength();
        }

        String startPosition = null;
        String endPosition = null;
        if (!CollectionUtils.isEmpty(message.getEntries())) {
            startPosition = buildPositionForDump(message.getEntries().get(0));
            endPosition = buildPositionForDump(message.getEntries().get(message.getEntries().size() - 1));
        }

        SimpleDateFormat format = new SimpleDateFormat(DATE_FORMAT);
        log.debug(context_format,
                new Object[] { batchId, size, memsize, format.format(new Date()), startPosition, endPosition });
    }

    protected void printColumn(List<CanalEntry.Column> columns) {
        for (CanalEntry.Column column : columns) {
            StringBuilder builder = new StringBuilder();
            builder.append(column.getName() + " : " + column.getValue());
            builder.append("    type=" + column.getMysqlType());
            if (column.getUpdated()) {
                builder.append("    update=" + column.getUpdated());
            }
            // builder.append(SEP);
            log.debug(builder.toString());
        }
    }

    protected void printEntry(List<CanalEntry.Entry> entrys) {
        for (CanalEntry.Entry entry : entrys) {
            long executeTime = entry.getHeader().getExecuteTime();
            long delayTime = new Date().getTime() - executeTime;

            log.debug("EntryType={}", entry.getEntryType());
            if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                CanalEntry.RowChange rowChage = null;
                try {
                    rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                } catch (Exception e) {
                    throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
                }

                CanalEntry.EventType eventType = rowChage.getEventType();
                String tabelName = entry.getHeader().getTableName();
                String schemaName = entry.getHeader().getSchemaName();

                log.debug(row_format,
                        new Object[] {
                                entry.getHeader().getLogfileName(),
                                String.valueOf(entry.getHeader().getLogfileOffset()),
                                schemaName,
                                tabelName,
                                eventType,
                                String.valueOf(entry.getHeader().getExecuteTime()),
                                String.valueOf(delayTime)
                        }
                );

                if (eventType == CanalEntry.EventType.QUERY || rowChage.getIsDdl()) {
                    log.debug(" sql ----> " + rowChage.getSql() + SEP);
                    continue;
                }
                try {
                    for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
                        if (eventType == CanalEntry.EventType.DELETE) {
                            printColumn(rowData.getBeforeColumnsList());
                            handlerColumn(tabelName, eventType, rowData.getBeforeColumnsList(), null);
                        } else if (eventType == CanalEntry.EventType.INSERT) {
                            printColumn(rowData.getAfterColumnsList());
                            handlerColumn(tabelName, eventType, null, rowData.getAfterColumnsList());
                        } else {
                            printColumn(rowData.getAfterColumnsList());
                            handlerColumn(tabelName, eventType, rowData.getBeforeColumnsList(), rowData.getAfterColumnsList());
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    log.error("Exception={}", ExceptionUtils.getFullStackTrace(e));
                }
            }
        }
    }

    /**
     * 处理信息
     * @param tabelName
     * @param eventType
     * @param beforeColumnsList
     * @param afterColumnsList
     * @throws Exception
     */
    private void handlerColumn(String tabelName, CanalEntry.EventType eventType, List<CanalEntry.Column> beforeColumnsList, List<CanalEntry.Column> afterColumnsList) throws Exception {

		String tableName = tabelName.toLowerCase();
		// topic目前规则为 ad-表名-cache
		String topicName = "ad-"+tableName+"-cache";

		// 判断是否包含对应的路由key
		if(Config.RABBIT_LZMH_ROUTING_KEY_MAP.get(tableName) != null){
			// 获取行变更信息 json
			String jsonStr = getRowChangeInfo(eventType, beforeColumnsList, afterColumnsList);

			// 发送信息 到rabbitMQ
			kafkaProducer.sendMsgToTopic(topicName,jsonStr);
			log.info("tabelName={} eventType={} jsonStr={}", tabelName, eventType, jsonStr);
		    // 发送map信息
            if(tabelName.equals("ssp_media_dev")){
                kafkaProducer.sendMsgToTopic("ad-"+tableName+"-mapcheck",jsonStr);
                log.info("tabelName={} eventType={} jsonStr={}", tabelName, eventType, jsonStr);
            }
		}

    }

    /**
     * 获取行的变更信息
     * @param eventType 事件类型
     * @param beforeColumnsList 变更前Columns
     * @param afterColumnsList 变更后 Columns
     * @return RowChangeInfo to Json
     */
    private String getRowChangeInfo(CanalEntry.EventType eventType, List<CanalEntry.Column> beforeColumnsList, List<CanalEntry.Column> afterColumnsList){

        // 行字段变更信息
        RowChangeInfo rowChangeInfo = new RowChangeInfo();

        // 变更字段，用“,”拼接
        String changeColumns = "";
        // 是否有变更字段
        boolean isUpdate = false;

        // 判断是否为update
        if (CanalEntry.EventType.UPDATE == eventType && afterColumnsList != null && afterColumnsList.size() > 0) {

            // 变更字段
            Map<String, Object> columnMap = new HashMap<>();

            // 循环遍历字段
            for (CanalEntry.Column column : afterColumnsList) {
                if (StringUtils.isNotEmpty(column.getValue())) {
                    columnMap.put(column.getName(), column.getValue());
                }

                // 判断字段是否变更
                if (column.getUpdated()) {
                    changeColumns += StringUtils.isNotEmpty(changeColumns) ? "," + column.getName() : column.getName();
                    isUpdate = true;
                }
            }

            rowChangeInfo.setAfter(GSON.toJson(columnMap));
            rowChangeInfo.setBefore(getJsonStr(beforeColumnsList));
        } else { // insert 、delete、 no afterColumnsList
            rowChangeInfo.setAfter(getJsonStr(afterColumnsList));
            rowChangeInfo.setBefore(getJsonStr(beforeColumnsList));
        }

        rowChangeInfo.setChangeColumn(changeColumns);
        rowChangeInfo.setIsUpdate(isUpdate);

        String eventTypeStr = eventType == null ? "type_null" :  eventType.toString().toLowerCase();
        rowChangeInfo.setEventType(eventTypeStr);

        return GSON.toJson(rowChangeInfo);
    }

    /**
     * 将Column List 转出 json字符串
     * @param columnsList
     * @return
     */
    private String getJsonStr(List<CanalEntry.Column> columnsList){
        if (columnsList == null || columnsList.size() < 1) {
            return "";
        }

        Map<String, Object> columnMap = new HashMap<>();
        for (CanalEntry.Column column : columnsList) {
            // 剔除空字符或null的值，以防gson解析成对象的时候出现空值解析失败
            if (StringUtils.isNotEmpty(column.getValue())) {
                columnMap.put(column.getName(), column.getValue());
            }
        }
        return GSON.toJson(columnMap);
    }

    @Override
    public void destroy() throws Exception {
        log.info("Canal Processor destroy...");
        try {
            log.info("## stop the canal client");
            if (!running) {
                return;
            }
            running = false;
            if (thread != null) {
                try {
                    thread.join();
                } catch (InterruptedException e) {
                    // ignore
                }
            }
            MDC.remove("destination");
        } catch (Throwable e) {
            log.warn("##something goes wrong when stopping canal:\n{}", ExceptionUtils.getFullStackTrace(e));
        } finally {
            log.info("## canal client is down.");
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        log.info("Canal Processor init...");
    }

}
